package com.myorg.util;

import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.primefaces.model.TreeNode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.myorg.domain.BlockInfo;

public class HdfsHelper {

	final Logger logger = LoggerFactory.getLogger(HdfsHelper.class);
	private Configuration config = null;

	public HdfsHelper() {
		getConfig();
	}

	public List<BlockInfo> getBlockInfos(String source) throws IOException {
		Configuration conf = getConfig();

		FileSystem fileSystem = FileSystem.get(conf);
		Path srcPath = new Path(source);

		// Check if the file already exists
		if (!(ifExists(srcPath))) {
			logger.info("No such destination " + srcPath);
			return null;
		}
		// Get the filename out of the file path

		FileStatus fileStatus = fileSystem.getFileStatus(srcPath);

		BlockLocation[] blkLocations = fileSystem.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());

		fileSystem.close();

		List<BlockInfo> rtn = new ArrayList<BlockInfo>();
		for (BlockLocation blkLocation : blkLocations) {
			rtn.add(new BlockInfo(source, blkLocation));
		}
		return rtn;
	}

	public boolean ifExists(Path source) throws IOException {
		Configuration config = getConfig();
		FileSystem hdfs = FileSystem.get(config);
		boolean isExists = hdfs.exists(source);
		return isExists;
	}

	public Configuration getConfig() {
		if (this.config == null) {
			Configuration config0 = new Configuration();
			config0.setBoolean("fs.hdfs.impl.disable.cache", true);
			config0.addResource(new Path("/usr/local/hadoop/etc/hadoop/core-site.xml"));
			config0.addResource(new Path("/usr/local/hadoop/etc/hadoop/hdfs-site.xml"));
			this.config = config0;
		}
		return this.config;
	}

	public static interface IHDFSFileTreeNodeProcessor {
		public TreeNode process(TreeNode parent, FileStatus fs, String startRootPath, String parentDir);
	}

	public void walkthrough(FileSystem fs, TreeNode parentNode, String parentDir, IHDFSFileTreeNodeProcessor processor,
			final String startRootPath) {
		try {
			for (FileStatus child : fs.listStatus(new Path(parentDir))) {
				TreeNode childNode = processor.process(parentNode, child, startRootPath, parentDir);
				if (childNode != null && child.isDirectory()) {
					walkthrough(fs, childNode, ("/".equals(parentDir) ? "/" + child.getPath().getName() : parentDir
							+ "/" + child.getPath().getName()), processor, startRootPath);
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		}
	}

	public void removeFile(String targetPath) {
		Configuration conf = getConfig();
		FileSystem fs = null;
		try {
			fs = FileSystem.get(URI.create(targetPath), conf);
			Path p = new Path(targetPath);
			fs.delete(p, true);
		} catch (IllegalArgumentException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		} catch (IOException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		} finally {
			if (fs != null) {
				try {
					fs.close();
				} catch (IOException e) {
				}
			}
		}
	}

	public void saveToHDFS(String targetPath, final InputStream in) {
		saveToHDFS(targetPath, new IHDFSFileSaver() {
			@Override
			public void save(final FSDataOutputStream targetOutputStream) throws IOException {
				IOUtils.copyBytes(in, targetOutputStream, 1024 * 1024);
			}
		});
	}

	public static interface IHDFSFileSaver {
		public void save(FSDataOutputStream targetOutputStream) throws IOException;
	}

	public void saveToHDFS(String targetPath, IHDFSFileSaver fileSaver) {
		Configuration conf = getConfig();
		FileSystem fs = null;
		FSDataOutputStream out = null;
		try {
			fs = FileSystem.get(URI.create(targetPath), conf);
			out = fs.create(new Path(targetPath));
			fileSaver.save(out);
		} catch (IllegalArgumentException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		} catch (IOException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		} finally {
			if (out != null) {
				IOUtils.closeStream(out);
			}
			if (fs != null) {
				try {
					fs.close();
				} catch (IOException e) {
				}
			}
		}
	}
}
